package ex.tableapi.join;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.$;

public class ExUnion extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExUnion demo = new ExUnion();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        Table orders1 = demo.registerMysql2("myorder1", "orders1");

        Table result = demo.exUnion(orders,orders1);//TODO

        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }

    //The UNION operation on two unbounded tables is currently not supported.
    private Table exUnion(Table orders, Table order1) {
//        Table left = tableEnv.from("orders1");
//        Table right = tableEnv.from("orders2");
        Table left = orders.where($("a").isEqual("liji"));
        Table right = order1.where($("a").isEqual("zy"));
        Table union = left.union(right);


        // 下面不能转换，只有流式表可以转成流
        // 出现异常：The UNION operation on two unbounded tables is currently not supported.
        //      DataStream<Tuple2<Boolean, Order>> resultDS = tenv.toRetractStream(result, Order.class);
        //      resultDS.print();

//        left.unionAll(right);
//        left.intersectAll(right);
//        left.intersect(right);
//        left.minus(right);
//        left.minusAll(right);
//        Table result = left.select($("a"), $("b"), $("c")).where($("a").in(right));
        return union;
    }


}
